package com.wchen.project.zk.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Desc: 使用 Curator 的异步接口
 *          1.前后两次使用了异步捷库 inBackground 来创建节点，一次返回是 0 ，另一次返回的是 -110，表明该节点已经存在
 *          2.两次分别调用 inBackground 接口时传入的 Executor 参数，第一次传入了 ExecutorService ，当收到异步通知
 *          事件时，相关异步事件处理逻辑就会交由该线程去做，第二次调用没有传入 Executor ，因此会使用 Zookeeper 默认的
 *          EventThread 来处理
 *
 *          为何要使用一个 Executor 来处理异步请求：在 Zk 中，所有异步通知事件都是由 EventThread 这个线程来处理的
 *          EventThread 线程用于串行处理所有的事件通知，目的是保证在这种“串行处理机制”下能够保证绝大部分场景下的
 *          事件执行的顺序性，但是这肯定也有弊端，就是一旦有某些事件处理单元过于复杂，就会影响到对其他事件的及时处理，
 *          所以，在 inBackground 接口中，允许用户传入一个 Executor 实例，这样就可以将那些比较负责的事件处理放到一个
 *          线程池中去处理，保证其他事件处理不受影响
 * @author: cww
 * @DateTime: 2020-01-24 15:08
 */

public class ZkCreateNodeByBackground {
    static String path = "/zk-book/c1";
    static CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString("home:2181")
            .sessionTimeoutMs(5000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .build();
    static CountDownLatch semaphore = new CountDownLatch(2);
    static ExecutorService tp = Executors.newFixedThreadPool(2);

    public static void main(String[] args) throws Exception{
        client.start();
        System.out.println("Main thread: " + Thread.currentThread().getName());
        // 此处传入了自定义的 Executor
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                System.out.println("event[code: " +event.getResultCode() + ", type: " + event.getType() + "]");
                System.out.println("Thread of processResult: " + Thread.currentThread().getName());
                semaphore.countDown();
            }
        }, tp).forPath(path, "init".getBytes());

        // 此处没有传入自定义的 Executor
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                System.out.println("event[code: " +event.getResultCode() + ", type: " + event.getType() + "]");
                System.out.println("Thread of processResult: " + Thread.currentThread().getName());
                semaphore.countDown();
            }
        }).forPath(path, "init".getBytes());
        semaphore.await();
        tp.shutdown();
    }
}
